package com.spi.mqtrans;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.spi.mapper.TransMapper;
import com.spi.mapper.UserMapper;
import com.spi.model.Trans;
import com.spi.model.User;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.HashMap;
import java.util.List;

@Component("mqTransactionListener")
@Slf4j
public class MqTransactionListener implements TransactionListener {

//    @Autowired
//    private UserMapper userMapper ;
//
//    @Autowired
//    private MqMessageDtoMapper mqMessageDtoMapper ;

    @Autowired
    private TransMapper transMapper ;

    @Autowired
    private UserMapper userMapper ;

    /**
     * 半事务消息发送成功后回调此方法 执行本地事务
     * @param message
     * @param o
     * @return
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {

        Trans trans = new Trans();
        LocalTransactionState state ; //用于告知mq二次确认结果
        String transactionId = null ;
        String messageid = null ;
        try {
            transactionId = message.getTransactionId();
            log.info("事务id为：{}",transactionId);
            String mess = new String(message.getBody());
            log.info("开始执行本地事务：用户信息：" + mess);
            //User user = JSONObject.parseObject(mess, User.class);
            HashMap hashMap = JSONObject.parseObject(mess, HashMap.class);
            User user = JSONObject.parseObject((String) hashMap.get("user"), User.class);
            messageid = (String)hashMap.get("messageid"); //消息id
            //1:幂等性判断  必须进行幂等性判断
            LambdaQueryWrapper<Trans> lambdaQueryWrapper = new LambdaQueryWrapper<>();
            lambdaQueryWrapper.eq(Trans::getTransid,transactionId)
                    .eq(Trans::getMessagekey,messageid)
                    .eq(Trans::getProducerstatus,1);
            List<Trans> trans1 = transMapper.selectList(lambdaQueryWrapper);
            if (trans1 != null && trans1.size() > 0 ){
                state = LocalTransactionState.COMMIT_MESSAGE; //提交
                return state ;
            }
            //2:执行业务逻辑
            LambdaQueryWrapper<User> lambdaQueryWrapper1 = new LambdaQueryWrapper<>();
            lambdaQueryWrapper1.eq(User::getUserid,user.getUserid());
            int update = userMapper.update(user, lambdaQueryWrapper1);
            if (update != 1){
                throw new RuntimeException("更新员工状态失败。。。") ;
            }
            //3:事务消息落库
            trans.setTransid(transactionId);
            trans.setMessagekey(messageid);
            trans.setProducerstatus(1);
            trans.setCreatetime(new Date());
            trans.setConsumerstatus(0);
            transMapper.insert(trans) ;
            state = LocalTransactionState.COMMIT_MESSAGE; //提交
            //System.out.println(1/0);; //抛异常
        }catch (Exception e){
            log.info("本地事务执行异常：{}",e.getMessage());
            e.printStackTrace();
            state = LocalTransactionState.ROLLBACK_MESSAGE;//回滚
            throw new RuntimeException();
        }
        return state;
    }

    /**
     * 回查接口，unknow状态，业务方执行本地事务逻辑时间耗时过程或网络原因引起，会导致broker对事务消息进行回查
     * 默认回查此时15次，第一次回查间隔时间6s 后续每次间隔60s
     * @param messageExt
     * @return
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
        String messageid = null ;
        String transactionId = messageExt.getTransactionId();
        LocalTransactionState state ; //用于告知mq二次确认结果
        log.info("事务id为：{}",transactionId);
        try {
            String mess = new String(messageExt.getBody());
            HashMap hashMap = JSONObject.parseObject(mess, HashMap.class);
            User user = JSONObject.parseObject((String) hashMap.get("user"), User.class);
            messageid = (String)hashMap.get("messageid"); //消息id
            LambdaQueryWrapper<Trans> lambdaQueryWrapper = new LambdaQueryWrapper<>();
            lambdaQueryWrapper.eq(Trans::getTransid,transactionId)
                    .eq(Trans::getMessagekey,messageid);
                    //.eq(Trans::getProducerstatus,1);
            List<Trans> trans1 = transMapper.selectList(lambdaQueryWrapper);
            if (trans1 != null && trans1.size() > 0 ){
                if ("1".equals(trans1.get(0).getProducerstatus())){
                    state = LocalTransactionState.COMMIT_MESSAGE; //提交
                    return state ;
                }
                if ("2".equals(trans1.get(0).getProducerstatus())){
                    state = LocalTransactionState.ROLLBACK_MESSAGE;
                    return  state ;
                }
            }
            state = LocalTransactionState.UNKNOW ;
            return state ;

        }catch (Exception e){
            e.printStackTrace();
            throw new RuntimeException();
        }
    }
}
